#include "AsyncWorker.h"

namespace mylog
{
    AsyncWorker::AsyncWorker(const functor& callback, AsyncType asyncType)
        : callback_(callback)
        , async_type_(asyncType)
        , stop_(false)
        , worker_(std::thread(&AsyncWorker::consumer, this))
    {
    }

    AsyncWorker::~AsyncWorker()
    {
        stop();
    }

    void AsyncWorker::stop()
    {
        stop_ = true;
        // 所有线程把缓冲区内的数据处理完，虽然这里只有一个线程
        cond_consumer_.notify_all();
        worker_.join();
    }

    void AsyncWorker::push(const char* data, size_t len)
    {
        // 停止了则不再继续处理新数据
        if (stop_)
            return;
        // 如果生产者队列不足以写下len长度的数据，并且缓冲区固定大小
        // 那就只能等消费者处理完之后将重置的缓冲区变成生产者缓冲区
        std::unique_lock<std::mutex> lock(mtx_);
        if (AsyncType::ASYNC_SAFE == async_type_)
        {
            // lambda表达式是防止虚假唤醒，只有当lambda表达式返回true时才会真正唤醒
            cond_producer_.wait(lock, [&](){
                return len <= buffer_producer_.writableSize();
            });
        }
        buffer_producer_.push(data, len);
        // 加入一份数据就叫醒一个消费者
        cond_consumer_.notify_one();
    }

    void AsyncWorker::consumer()
    {
        while(1)
        {
            {
                // 交换生产者和消费者的缓冲区，这样消费者在消费时，生产者可以不用和消费者抢夺缓冲区
                std::unique_lock<std::mutex> lock(mtx_);
                // 当stop为false时，生产者缓冲区为空，则阻塞等待，直到有数据后被唤醒后进行对换后消费；否则进行对换后消费
                // 当stop为true时，生产者缓冲区为空，则直接返回；否则进行对换后消费
                if (buffer_producer_.isEmpty())
                {
                    if (stop_) 
                        return;
                    else 
                    {
                        cond_consumer_.wait(lock, [&](){
                            // 被唤醒时若生产者缓冲区非空，则继续往下执行
                            // 被唤醒时若stop_为true，说明需要停止，也可以继续往下执行
                            return !buffer_producer_.isEmpty() || stop_;
                        });
                    }
                }
                buffer_producer_.swap(buffer_consumer_);
                // 固定容量的话，由于有生产者被阻塞，因此需要主动唤醒生产者
                // 因为上次处理完成之后buffer_consumer_必然被重置有闲置空间，因此生产者必然有空间可以写入
                // 只通知一个是因为，一次只能有一个生产者往生产者线程写入，唤醒多个只有一个能写，其他还是要等下一轮唤醒
                if (async_type_ == AsyncType::ASYNC_SAFE) 
                    cond_producer_.notify_one();
            }
            callback_(buffer_consumer_);
            buffer_consumer_.reset();
            // 处理完原有数据，但是还是在停止之前有新数据加入，则继续处理
            // 因为可能有生产者线程阻塞，这次对换处理之后刚好来了新数据。
            if (stop_ && buffer_producer_.isEmpty()) 
                return;
        }
    }


} // namespace log